package com.red.wood.task;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import com.red.wood.model.SqlInfo;
import com.red.wood.util.CommonUtils;
import com.red.wood.util.SqlParser;
import com.red.wood.util.SysProps;

public class DomainVerifyBaseTask {

	private static final Log log = LogFactory.getLog(DomainVerifyBaseTask.class);
	
	public static void domainTask(SparkSession spark,SqlInfo sqlInfo,String day) {
		try {
			SysProps sysProps = SysProps.getInstance();
			
			String hive_databse = sysProps.getValue("hive.database");
			String domain_base_verify_hdfs = sysProps.getValue("domain_base_verify_hdfs");
			
			//DNS基础上报  
			dnsVerifyDnsserver(spark,sqlInfo,day,domain_base_verify_hdfs,hive_databse);
			
			//IDC活跃域名数据错误核验表
			idcActiveVerifyDomain(spark,sqlInfo,day,domain_base_verify_hdfs,hive_databse);
			
			//IDC基础域名数据错误核验表
			idcBasicDomainVerify(spark,sqlInfo,day,domain_base_verify_hdfs,hive_databse);
			
			//spark.sql("insert overwrite table "+hive_databse+".t_dm_domain_verify_base partition(ds='"+day+"') select domain,top_domain,dns_last_active_time,dns_visit_count,dns_reg_status,dns_reg_org,dns_reg_registrar,dns_reg_date,dns_reg_exp_date,dns_reg_updated_date,dns_auth_ts_open_status,dns_auth_ts_class,dns_auth_ts_hosted_start_time,idc_user_id,idc_user_name,idc_service_id,idc_service_content,idc_regtype,idc_license_key,idc_inject_type,idc_business,idc_find_time,idc_last_active_time,idc_is_block,idc_active_visit_count,ircs_user_id,ircs_user_name,ircs_service_id,ircs_service_content,ircs_regtype,ircs_license_key,ircs_inject_type,ircs_find_time,ircs_last_active_time,ircs_is_block,ircs_active_visit_count,cdn_user_id,cdn_user_name,cdn_service_id,cdn_source,cdn_license_key,cdn_find_time,cdn_active_visit_count,cdn_data_source,is_cdn,icp_wzmc,icp_nrlx,icp_fwnr,icp_license_key,icp_state,icp_detect_time,domain_suffix,domain_level,ip_countrys,ip_provinces,ip_citys,ip_towns,idc_isp_ids,idc_isp_names,house_codes,house_names,operators_ids,operators_names,is_zx,is_dns_parse,is_alive,dail_time,survival_time,crawler_find_time,data_source,update_time,verify_type from "+hive_databse+".t_dm_domain_verify_base where 1 = 2");
			long count = spark.sql("select ds from "+hive_databse+".t_dm_domain_verify_base where ds='"+day+"' limit 1").count();
			if(count > 0) {
				spark.sql("truncate table "+hive_databse+".t_dm_domain_verify_base partition(ds='"+day+"')");
			}
			String dwdPath= domain_base_verify_hdfs+"/*";
			String sql ="load data inpath '"+dwdPath+"/*' into table "+hive_databse+".t_dm_domain_verify_base partition (ds='"+day+"')";//overwrite
			
			spark.sql(sql);
		}catch(Exception e) {
			if(log.isErrorEnabled()) {
				log.error(e.getMessage(),e);
			}
		}
	}
	
	private static void idcBasicDomainVerify(SparkSession spark,SqlInfo sqlInfo,String day,String domain_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcBasicDomainVerify = sqlInfo.getSqlIdcBasicDomainVerify();
		sqlIdcBasicDomainVerify = SqlParser.getOriginalOutputSql(sqlIdcBasicDomainVerify,"hive_databse",hive_databse);
		sqlIdcBasicDomainVerify = SqlParser.getSql(sqlIdcBasicDomainVerify, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcBasicDomainVerify);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String domain = CommonUtils.valueOf(row.getAs("domain"));
					String top_domain = "";
					String dns_last_active_time = "";
					String dns_visit_count = "";
					String dns_reg_status = "";
					String dns_reg_org = "";
					String dns_reg_registrar = "";
					String dns_reg_date = "";
					String dns_reg_exp_date = "";
					String dns_reg_updated_date = "";
					String dns_auth_ts_open_status = "";
					String dns_auth_ts_class = "";
					String dns_auth_ts_hosted_start_time = "";
					String idc_user_id = CommonUtils.valueOf(row.getAs("idc_user_id"));
					String idc_user_name = CommonUtils.valueOf(row.getAs("idc_user_name"));
					String idc_service_id = CommonUtils.valueOf(row.getAs("idc_service_id"));
					String idc_service_content = "";
					String idc_regtype = "";
					String idc_license_key = "";
					String idc_inject_type = "";
					String idc_business = "";
					String idc_find_time = "";
					String idc_last_active_time = "";
					String idc_is_block = "";
					String idc_active_visit_count = "";
					String ircs_user_id = "";
					String ircs_user_name = "";
					String ircs_service_id = "";
					String ircs_service_content = "";
					String ircs_regtype = "";
					String ircs_license_key = "";
					String ircs_inject_type = "";
					String ircs_find_time = "";
					String ircs_last_active_time = "";
					String ircs_is_block = "";
					String ircs_active_visit_count = "";
					String cdn_user_id = "";
					String cdn_user_name = "";
					String cdn_service_id = "";
					String cdn_source = "";
					String cdn_license_key = "";
					String cdn_find_time = "";
					String cdn_active_visit_count = "";
					String cdn_data_source = "";
					String is_cdn = "";
					String icp_wzmc = "";
					String icp_nrlx = "";
					String icp_fwnr = "";
					String icp_license_key = "";
					String icp_state = "";
					String icp_detect_time = "";
					String domain_suffix = "";
					String domain_level = "";
					String ip_countrys = "";
					String ip_provinces = CommonUtils.valueOf(row.getAs("ip_provinces"));
					String ip_citys = "";
					String ip_towns = "";
					String idc_isp_ids = CommonUtils.valueOf(row.getAs("idc_isp_ids"));
					String idc_isp_names = CommonUtils.valueOf(row.getAs("idc_isp_names"));
					String house_codes = "";
					String house_names = "";
					String operators_ids = "";
					String operators_names = "";
					String is_zx = "";
					String is_dns_parse = "";
					String is_alive = "";
					String dail_time = "";
					String survival_time = "";
					String crawler_find_time = "";
					String data_source = "";
					String update_time = CommonUtils.valueOf(row.getAs("update_time"));
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));	
					
					buffer.append(domain+"\t").append(top_domain+"\t").append(dns_last_active_time+"\t").append(dns_visit_count+"\t").append(dns_reg_status+"\t")
						  .append(dns_reg_org+"\t").append(dns_reg_registrar+"\t").append(dns_reg_date+"\t").append(dns_reg_exp_date+"\t").append(dns_reg_updated_date+"\t")
						  .append(dns_auth_ts_open_status+"\t").append(dns_auth_ts_class+"\t").append(dns_auth_ts_hosted_start_time+"\t").append(idc_user_id+"\t").append(idc_user_name+"\t")
						  .append(idc_service_id+"\t").append(idc_service_content+"\t").append(idc_regtype+"\t").append(idc_license_key+"\t").append(idc_inject_type+"\t")
						  .append(idc_business+"\t").append(idc_find_time+"\t").append(idc_last_active_time+"\t").append(idc_is_block+"\t").append(idc_active_visit_count+"\t")
						  .append(ircs_user_id+"\t").append(ircs_user_name+"\t").append(ircs_service_id+"\t").append(ircs_service_content+"\t").append(ircs_regtype+"\t")
						  .append(ircs_license_key+"\t").append(ircs_inject_type+"\t").append(ircs_find_time+"\t").append(ircs_last_active_time+"\t").append(ircs_is_block+"\t")
						  .append(ircs_active_visit_count+"\t").append(cdn_user_id+"\t").append(cdn_user_name+"\t").append(cdn_service_id+"\t").append(cdn_source+"\t")
						  .append(cdn_license_key+"\t").append(cdn_find_time+"\t").append(cdn_active_visit_count+"\t").append(cdn_data_source+"\t").append(is_cdn+"\t")
						  .append(icp_wzmc+"\t").append(icp_nrlx+"\t").append(icp_fwnr+"\t").append(icp_license_key+"\t").append(icp_state+"\t")
						  .append(icp_detect_time+"\t").append(domain_suffix+"\t").append(domain_level+"\t").append(ip_countrys+"\t").append(ip_provinces+"\t")
						  .append(ip_citys+"\t").append(ip_towns+"\t").append(idc_isp_ids+"\t").append(idc_isp_names+"\t").append(house_codes+"\t")
						  .append(house_names+"\t").append(operators_ids+"\t").append(operators_names+"\t").append(is_zx+"\t").append(is_dns_parse+"\t")
						  .append(is_alive+"\t").append(dail_time+"\t").append(survival_time+"\t").append(crawler_find_time+"\t").append(data_source+"\t")
						  .append(update_time+"\t").append(verify_type+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= domain_base_verify_hdfs+"/idcBasicDomainVerify";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}	
	
	private static void idcActiveVerifyDomain(SparkSession spark,SqlInfo sqlInfo,String day,String domain_base_verify_hdfs,String hive_databse) {
		
		String sqlIdcActiveVerifyDomain = sqlInfo.getSqlIdcActiveVerifyDomain();
		sqlIdcActiveVerifyDomain = SqlParser.getOriginalOutputSql(sqlIdcActiveVerifyDomain,"hive_databse",hive_databse);
		sqlIdcActiveVerifyDomain = SqlParser.getSql(sqlIdcActiveVerifyDomain, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlIdcActiveVerifyDomain);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String domain = CommonUtils.valueOf(row.getAs("domain"));
					String top_domain = CommonUtils.valueOf(row.getAs("top_domain"));
					String dns_last_active_time = "";
					String dns_visit_count = "";
					String dns_reg_status = "";
					String dns_reg_org = "";
					String dns_reg_registrar = "";
					String dns_reg_date = "";
					String dns_reg_exp_date = "";
					String dns_reg_updated_date = "";
					String dns_auth_ts_open_status = "";
					String dns_auth_ts_class = "";
					String dns_auth_ts_hosted_start_time = "";
					String idc_user_id = "";
					String idc_user_name = "";
					String idc_service_id = "";
					String idc_service_content = "";
					String idc_regtype = "";
					String idc_license_key = "";
					String idc_inject_type = "";
					String idc_business = "";
					String idc_find_time = CommonUtils.valueOf(row.getAs("idc_find_time"));
					String idc_last_active_time = CommonUtils.valueOf(row.getAs("idc_last_active_time"));
					String idc_is_block = CommonUtils.valueOf(row.getAs("idc_is_block"));
					String idc_active_visit_count = CommonUtils.valueOf(row.getAs("idc_active_visit_count"));
					String ircs_user_id = "";
					String ircs_user_name = "";
					String ircs_service_id = "";
					String ircs_service_content = "";
					String ircs_regtype = "";
					String ircs_license_key = "";
					String ircs_inject_type = "";
					String ircs_find_time = "";
					String ircs_last_active_time = "";
					String ircs_is_block = "";
					String ircs_active_visit_count = "";
					String cdn_user_id = "";
					String cdn_user_name = "";
					String cdn_service_id = "";
					String cdn_source = "";
					String cdn_license_key = "";
					String cdn_find_time = "";
					String cdn_active_visit_count = "";
					String cdn_data_source = "";
					String is_cdn = "";
					String icp_wzmc = "";
					String icp_nrlx = "";
					String icp_fwnr = "";
					String icp_license_key = "";
					String icp_state = "";
					String icp_detect_time = "";
					String domain_suffix = "";
					String domain_level = "";
					String ip_countrys = "";
					String ip_provinces = CommonUtils.valueOf(row.getAs("ip_provinces"));
					String ip_citys = "";
					String ip_towns = "";
					String idc_isp_ids = CommonUtils.valueOf(row.getAs("idc_isp_ids"));
					String idc_isp_names = CommonUtils.valueOf(row.getAs("idc_isp_names"));
					String house_codes = CommonUtils.valueOf(row.getAs("house_codes"));
					String house_names = "";
					String operators_ids = "";
					String operators_names = "";
					String is_zx = "";
					String is_dns_parse = "";
					String is_alive = "";
					String dail_time = "";
					String survival_time = "";
					String crawler_find_time = "";
					String data_source = "";
					String update_time = CommonUtils.valueOf(row.getAs("update_time"));
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));	
					
					buffer.append(domain+"\t").append(top_domain+"\t").append(dns_last_active_time+"\t").append(dns_visit_count+"\t").append(dns_reg_status+"\t")
						  .append(dns_reg_org+"\t").append(dns_reg_registrar+"\t").append(dns_reg_date+"\t").append(dns_reg_exp_date+"\t").append(dns_reg_updated_date+"\t")
						  .append(dns_auth_ts_open_status+"\t").append(dns_auth_ts_class+"\t").append(dns_auth_ts_hosted_start_time+"\t").append(idc_user_id+"\t").append(idc_user_name+"\t")
						  .append(idc_service_id+"\t").append(idc_service_content+"\t").append(idc_regtype+"\t").append(idc_license_key+"\t").append(idc_inject_type+"\t")
						  .append(idc_business+"\t").append(idc_find_time+"\t").append(idc_last_active_time+"\t").append(idc_is_block+"\t").append(idc_active_visit_count+"\t")
						  .append(ircs_user_id+"\t").append(ircs_user_name+"\t").append(ircs_service_id+"\t").append(ircs_service_content+"\t").append(ircs_regtype+"\t")
						  .append(ircs_license_key+"\t").append(ircs_inject_type+"\t").append(ircs_find_time+"\t").append(ircs_last_active_time+"\t").append(ircs_is_block+"\t")
						  .append(ircs_active_visit_count+"\t").append(cdn_user_id+"\t").append(cdn_user_name+"\t").append(cdn_service_id+"\t").append(cdn_source+"\t")
						  .append(cdn_license_key+"\t").append(cdn_find_time+"\t").append(cdn_active_visit_count+"\t").append(cdn_data_source+"\t").append(is_cdn+"\t")
						  .append(icp_wzmc+"\t").append(icp_nrlx+"\t").append(icp_fwnr+"\t").append(icp_license_key+"\t").append(icp_state+"\t")
						  .append(icp_detect_time+"\t").append(domain_suffix+"\t").append(domain_level+"\t").append(ip_countrys+"\t").append(ip_provinces+"\t")
						  .append(ip_citys+"\t").append(ip_towns+"\t").append(idc_isp_ids+"\t").append(idc_isp_names+"\t").append(house_codes+"\t")
						  .append(house_names+"\t").append(operators_ids+"\t").append(operators_names+"\t").append(is_zx+"\t").append(is_dns_parse+"\t")
						  .append(is_alive+"\t").append(dail_time+"\t").append(survival_time+"\t").append(crawler_find_time+"\t").append(data_source+"\t")
						  .append(update_time+"\t").append(verify_type+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= domain_base_verify_hdfs+"/idcActiveVerifyDomain";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}	
	
	private static void dnsVerifyDnsserver(SparkSession spark,SqlInfo sqlInfo,String day,String domain_base_verify_hdfs,String hive_databse) {
		
		String sqlDnsVerifyDnsserver = sqlInfo.getSqlDnsVerifyDnsserver();
		sqlDnsVerifyDnsserver = SqlParser.getOriginalOutputSql(sqlDnsVerifyDnsserver,"hive_databse",hive_databse);
		sqlDnsVerifyDnsserver = SqlParser.getSql(sqlDnsVerifyDnsserver, "query_day", day);
		
		Dataset<Row> dataset = spark.sql(sqlDnsVerifyDnsserver);
		
		JavaRDD<String> rdd = dataset.toJavaRDD().map(new Function<Row, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public String call(Row row){
				StringBuffer buffer = new StringBuffer();
				try {
					String domain = CommonUtils.valueOf(row.getAs("domain"));
					String top_domain = "";
					String dns_last_active_time = "";
					String dns_visit_count = "";
					String dns_reg_status = "";
					String dns_reg_org = "";
					String dns_reg_registrar = "";
					String dns_reg_date = CommonUtils.valueOf(row.getAs("dns_reg_date"));
					String dns_reg_exp_date = "";
					String dns_reg_updated_date = "";
					String dns_auth_ts_open_status = "";
					String dns_auth_ts_class = "";
					String dns_auth_ts_hosted_start_time = "";
					String idc_user_id = "";
					String idc_user_name = "";
					String idc_service_id = "";
					String idc_service_content = "";
					String idc_regtype = "";
					String idc_license_key = "";
					String idc_inject_type = "";
					String idc_business = "";
					String idc_find_time = "";
					String idc_last_active_time = "";
					String idc_is_block = "";
					String idc_active_visit_count = "";
					String ircs_user_id = "";
					String ircs_user_name = "";
					String ircs_service_id = "";
					String ircs_service_content = "";
					String ircs_regtype = "";
					String ircs_license_key = "";
					String ircs_inject_type = "";
					String ircs_find_time = "";
					String ircs_last_active_time = "";
					String ircs_is_block = "";
					String ircs_active_visit_count = "";
					String cdn_user_id = "";
					String cdn_user_name = "";
					String cdn_service_id = "";
					String cdn_source = "";
					String cdn_license_key = "";
					String cdn_find_time = "";
					String cdn_active_visit_count = "";
					String cdn_data_source = "";
					String is_cdn = "";
					String icp_wzmc = "";
					String icp_nrlx = "";
					String icp_fwnr = "";
					String icp_license_key = "";
					String icp_state = "";
					String icp_detect_time = "";
					String domain_suffix = "";
					String domain_level = "";
					String ip_countrys = "";
					String ip_provinces = CommonUtils.valueOf(row.getAs("ip_provinces"));
					String ip_citys = "";
					String ip_towns = "";
					String idc_isp_ids = "";
					String idc_isp_names = "";
					String house_codes = CommonUtils.valueOf(row.getAs("house_codes"));
					String house_names = CommonUtils.valueOf(row.getAs("house_names"));
					String operators_ids = "";
					String operators_names = "";
					String is_zx = "";
					String is_dns_parse = "";
					String is_alive = "";
					String dail_time = "";
					String survival_time = "";
					String crawler_find_time = "";
					String data_source = "";
					String update_time = "";
					String verify_type = CommonUtils.valueOf(row.getAs("verify_type"));	
					
					buffer.append(domain+"\t").append(top_domain+"\t").append(dns_last_active_time+"\t").append(dns_visit_count+"\t").append(dns_reg_status+"\t")
						  .append(dns_reg_org+"\t").append(dns_reg_registrar+"\t").append(dns_reg_date+"\t").append(dns_reg_exp_date+"\t").append(dns_reg_updated_date+"\t")
						  .append(dns_auth_ts_open_status+"\t").append(dns_auth_ts_class+"\t").append(dns_auth_ts_hosted_start_time+"\t").append(idc_user_id+"\t").append(idc_user_name+"\t")
						  .append(idc_service_id+"\t").append(idc_service_content+"\t").append(idc_regtype+"\t").append(idc_license_key+"\t").append(idc_inject_type+"\t")
						  .append(idc_business+"\t").append(idc_find_time+"\t").append(idc_last_active_time+"\t").append(idc_is_block+"\t").append(idc_active_visit_count+"\t")
						  .append(ircs_user_id+"\t").append(ircs_user_name+"\t").append(ircs_service_id+"\t").append(ircs_service_content+"\t").append(ircs_regtype+"\t")
						  .append(ircs_license_key+"\t").append(ircs_inject_type+"\t").append(ircs_find_time+"\t").append(ircs_last_active_time+"\t").append(ircs_is_block+"\t")
						  .append(ircs_active_visit_count+"\t").append(cdn_user_id+"\t").append(cdn_user_name+"\t").append(cdn_service_id+"\t").append(cdn_source+"\t")
						  .append(cdn_license_key+"\t").append(cdn_find_time+"\t").append(cdn_active_visit_count+"\t").append(cdn_data_source+"\t").append(is_cdn+"\t")
						  .append(icp_wzmc+"\t").append(icp_nrlx+"\t").append(icp_fwnr+"\t").append(icp_license_key+"\t").append(icp_state+"\t")
						  .append(icp_detect_time+"\t").append(domain_suffix+"\t").append(domain_level+"\t").append(ip_countrys+"\t").append(ip_provinces+"\t")
						  .append(ip_citys+"\t").append(ip_towns+"\t").append(idc_isp_ids+"\t").append(idc_isp_names+"\t").append(house_codes+"\t")
						  .append(house_names+"\t").append(operators_ids+"\t").append(operators_names+"\t").append(is_zx+"\t").append(is_dns_parse+"\t")
						  .append(is_alive+"\t").append(dail_time+"\t").append(survival_time+"\t").append(crawler_find_time+"\t").append(data_source+"\t")
						  .append(update_time+"\t").append(verify_type+"\t");
				}catch(Exception e) {
					if(log.isErrorEnabled()) {
						log.error(e.getMessage(),e);
					}
				}
				return buffer.toString();
			}
		});
		
		if(rdd.count() > 0) {
			String dwdPath= domain_base_verify_hdfs+"/dnsVerifyDnsserver";
			CommonUtils.uploadHdfs(rdd,dwdPath);
		}
	}	
}